package com.haizhi.util;


import org.apache.http.HttpHeaders;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.nio.file.Paths;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
 * 
 * 多线程的断点下载程序，根据输入的url和指定线程数，来完成断点续传功能。
 * 
 * 每个线程支负责某一小段的数据下载；再通过RandomAccessFile完成数据的整合。
 */
public class Downloader {

    private static Logger logger = LoggerFactory.getLogger(Downloader.class);

    private static final int BUFFER_SIZE = 1024 * 8;

    private static final int MAX_THREAD_NUM = 4;

    private String fileUrl = null;
    private String filePath = null;

    private long fileLength = 0L;
    private long partLength = 0L;

    private URL url = null;

    //有参构造函数，先构造需要的数据
    public Downloader(String fileUrl, String filePath) {
        this.fileUrl = fileUrl;
        this.filePath = filePath;

        //创建文件夹
        Tools.createDirs(filePath);
    }

    //获得线程数目
    private int getThreadNum(long fileLength) {

        int threadNum = 1;

        if (fileLength <= BUFFER_SIZE) {
            return threadNum;
        }

        threadNum = (int) Math.ceil(fileLength / BUFFER_SIZE);

        if (threadNum >= MAX_THREAD_NUM) {
            threadNum = MAX_THREAD_NUM;
        }

        return threadNum;
    }

    /*
     * 组织断点续传功能的方法
     */
    public boolean downloadPart() {
        boolean ret;
        int threadNum;
        File file;
        File tmp_file;
        HttpURLConnection httpURLConnection;

        //在请求url内获取文件资源的名称；此处没考虑文件名为空的情况，此种情况可能需使用UUID来生成一个唯一数来代表文件名。
        String tmp_filepath = filePath + "_tmp";
        long start = System.currentTimeMillis();
        do {
            try {
                url = new URL(fileUrl);
                httpURLConnection = (HttpURLConnection) url.openConnection();
                httpURLConnection.setRequestProperty(HttpHeaders.CONNECTION, "Close");
                fileLength = httpURLConnection.getContentLengthLong();//获取请求资源的总长度。
                logger.info("文件长度: length = " + fileLength);
                if (fileLength == -1) {
                    logger.error("获取文件长度失败, 可能网络异常: {}", filePath);
                    ret = false;
                    break;
                }
                httpURLConnection.disconnect();

                //根据文件大小获得相应的线程数目
                threadNum = getThreadNum(fileLength);
                logger.info("下载线程数目为: threadNum = {} ", threadNum);

                file = new File(filePath);
                tmp_file = new File(tmp_filepath);

                long[] startPos = new long[threadNum];
                long[] endPos = new long[threadNum];
                CountDownLatch latch = new CountDownLatch(threadNum);

                //每个线程需下载的资源大小。
                partLength = fileLength / threadNum;
                logger.info("filePath: " + filePath + " ," + "fileLength= "
                        + fileLength + " the partLength= " + partLength);

                if (file.exists() && !tmp_file.exists()) {
                    logger.info("the file you want to download has exited!!");
                    boolean flag = file.delete();
                    if (flag) {
                        logger.info("删除已存在文件成功: {}", filePath);
                    }
                }

                setBreakPoint(startPos, endPos, tmp_file, threadNum);
                ExecutorService exec = Executors.newCachedThreadPool();
                for (int i = 0; i < threadNum; i++) {
                    exec.execute(new DownLoadThread(startPos[i], endPos[i],
                            this, i, tmp_file, latch));
                }
                //当你的计数器减为0之前，会在此处一直阻塞。
                latch.await();
                exec.shutdown();

                logger.info("下载后的文件大小为: length = " + file.length());
                if (tmp_file.exists()) {
                    logger.info("下载完成后删除tmp文件: tmp = {}!!", tmp_filepath);
                    boolean flag = tmp_file.delete();
                    if (flag) {
                        logger.info("删除临时文件成功 {}", tmp_filepath);
                    }
                }
                ret = true;
                break;
            } catch (Exception e) {
                logger.error("ERROR:", e);
                ret = false;
                break;
            }
        } while (true);
        long end = System.currentTimeMillis();

        logger.info("{} {} 一共耗时: {}毫秒 {} kb/s", filePath, fileLength, (end - start), fileLength / 1024.0 / ((end - start) / 1000.0));
        return ret;
    }

    /*
     * 断点设置方法，当有临时文件时，直接在临时文件中读取上次下载中断时的断点位置。没有临时文件，即第一次下载时，重新设置断点。
     * 
     * rand_tmp_file.seek()跳转到一个位置的目的是为了让各个断点存储的位置尽量分开。
     * 
     * 这是实现断点续传的重要基础。
     */
    private void setBreakPoint(long[] startPos, long[] endPos, File tmp_file, int threadNum) {
        RandomAccessFile rand_tmp_file = null;
        try {
            if (tmp_file.exists()) {
                logger.info("{} exists the download has continued!!", tmp_file.getPath());
                rand_tmp_file = new RandomAccessFile(tmp_file, "rw");
                for (int i = 0; i < threadNum; i++) {
                    rand_tmp_file.seek(8 * i + 8);
                    startPos[i] = rand_tmp_file.readLong();

                    rand_tmp_file.seek(8 * (i + 1000) + 16);
                    endPos[i] = rand_tmp_file.readLong();

                    logger.info("the Array content in the exit file: ");
                    logger.info("the thread" + i + " startPos:" + startPos[i] + ", endPos: " + endPos[i]);
                }
            } else {
                logger.info("the {} is not exists and create it!!", tmp_file.getPath());
                rand_tmp_file = new RandomAccessFile(tmp_file, "rw");

                //最后一个线程的截止位置大小为请求资源的大小
                for (int i = 0; i < threadNum; i++) {
                    startPos[i] = partLength * i;
                    if (i == threadNum - 1) {
                        endPos[i] = fileLength - 1;
                    } else {
                        endPos[i] = partLength * (i + 1) - 1;
                    }

                    if (startPos[i] < 0) {
                        startPos[i] = 0;
                    }
                    if (endPos[i] < 0) {
                        endPos[i] = 0;
                    }

                    rand_tmp_file.seek(8 * i + 8);
                    rand_tmp_file.writeLong(startPos[i]);

                    rand_tmp_file.seek(8 * (i + 1000) + 16);
                    rand_tmp_file.writeLong(endPos[i]);

                    logger.info("the Array content: ");
                    logger.info("the thread" + (i + 1) + " startPos:" + startPos[i] + ", endPos: " + endPos[i]);
                }
            }
        } catch (Exception e) {
            logger.error("ERROR:", e);
        } finally {
            try {
                if (rand_tmp_file != null) {
                    rand_tmp_file.close();
                }
            } catch (IOException e) {
                logger.error("ERROR:", e);
            }
        }
    }

    /*
     * 实现下载功能的内部类，通过读取断点来设置向服务器请求的数据区间。
     */
    class DownLoadThread implements Runnable {

        private long startPos;
        private long endPos;
        private Downloader task = null;
        private RandomAccessFile download_file = null;
        private RandomAccessFile rand_tmp_file = null;
        private int threadId;
        private File tmp_file = null;
        private CountDownLatch latch = null;

        DownLoadThread(long startPos, long endPos,
                       Downloader task, int threadId,
                       File tmp_file, CountDownLatch latch) {
            this.startPos = startPos;
            this.endPos = endPos;
            this.task = task;
            this.tmp_file = tmp_file;
            this.threadId = threadId;
            this.latch = latch;
        }

        @Override
        public void run() {

            int length = 0;
            InputStream is = null;
            HttpURLConnection httpURLConnection = null;

            while (true) {
                try {
                    logger.info("the thread " + threadId + " has started!!");
                    try {
                        this.download_file = new RandomAccessFile(this.task.filePath, "rw");
                        this.rand_tmp_file = new RandomAccessFile(this.tmp_file, "rw");
                    } catch (FileNotFoundException e) {
                        logger.error("打开文件失败: {}", filePath);
                        logger.error("ERROR:", e);
                    }

                    httpURLConnection = (HttpURLConnection) task.url.openConnection();
                    setHeader(httpURLConnection);

                    //防止网络阻塞，设置指定的超时时间；单位都是ms。超过指定时间，就会抛出异常
                    httpURLConnection.setReadTimeout(40000);//读取数据的超时设置
                    httpURLConnection.setConnectTimeout(40000);//连接的超时设置

                    long count = 0L;
                    byte[] buf = new byte[BUFFER_SIZE];

                    rand_tmp_file.seek(8 * threadId + 8);
                    startPos = rand_tmp_file.readLong();
                    logger.info("线程{}开始下载起始位置: {} - {}", threadId, startPos, endPos);
                    if (startPos < endPos) {

                        //向服务器请求指定区间段的数据，这是实现断点续传的根本。
                        httpURLConnection.setRequestProperty("Range", "bytes=" + startPos + "-" + endPos);

                        logger.info("Thread " + threadId + " the total size:---- " + (endPos - startPos + 1));

                        download_file.seek(startPos);

//                        logger.info("{} status = {}", threadId, httpURLConnection.getResponseCode());
//                        logger.info("{} content-length = {}", threadId, httpURLConnection.getContentLength());

                        is = httpURLConnection.getInputStream();//获取服务器返回的资源流
                        while ((length = is.read(buf)) != -1) {
                            count += length;
                            download_file.write(buf, 0, length);

                            //不断更新每个线程下载资源的起始位置，并写入临时文件；为断点续传做准备
                            startPos += length;
                            rand_tmp_file.seek(8 * threadId + 8);
                            rand_tmp_file.writeLong(startPos);

                            //如果已经下载完成, 则主动退出线程
                            if (startPos >= endPos) {
                                logger.info("线程已经下载完成: {} start = {} end = {}", threadId, startPos, endPos);
                                break;
                            }
                        }
                        logger.info("the thread " + threadId + " total load count: " + count);
                    }

                    logger.info("thread = {} startPos = {} endPos = {}", threadId, startPos, endPos);
                    if (startPos >= endPos) {
                        latch.countDown();//计数器自减
                        logger.info("the thread " + threadId + " has done!!");
                        break;
                    }
                    logger.warn("线程 {} 数据读取 length = {} startPos = {} endPos = {}",
                            threadId, length, startPos, endPos);
                } catch (IOException e) {
                    logger.error("下载异常: {}", threadId);
                    logger.error("ERROR:", e);
                } finally {

                    try {
                        if (is != null) {
                            //关闭IO流
                            is.close();
                            is = null;
                        }
                        //关闭文件 防止资源泄漏
                        if (download_file != null) {
                            download_file.close();
                            download_file = null;
                        }

                        if (rand_tmp_file != null) {
                            rand_tmp_file.close();
                            rand_tmp_file = null;
                        }
                    } catch (IOException e) {
                        logger.error("ERROR:", e);
                    }

                    if (httpURLConnection != null) {
                        httpURLConnection.disconnect();
                        httpURLConnection = null;
                    }
                    logger.info("{} 资源释放成功...", threadId);
                }
            }
        }
    }

    /*
     * 为一个HttpURLConnection模拟请求头，伪装成一个浏览器发出的请求
     */
    private void setHeader(HttpURLConnection con) {
        con.setRequestProperty("User-Agent",
                "Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9.0.3) Gecko/2008092510 Ubuntu/8.04 (hardy) Firefox/3.0.3");
        con.setRequestProperty("Accept-Language", "en-us,en;q=0.7,zh-cn;q=0.3");
        con.setRequestProperty("Accept-Charset", "ISO-8859-1,utf-8;q=0.7,*;q=0.7");
        con.setRequestProperty("Keep-Alive", "300");
        con.setRequestProperty("Connection", "keep-alive");
    }

    public static void main(String[] args) {

        String fileUrl = "http://localhost:5588/download/spring4.pdf";
        String filePath = "spring4.pdf";
        Downloader load = new Downloader(fileUrl, filePath);
        load.downloadPart();

        String hash = "";
        try {
            hash = GenerateMd5.getHash(Paths.get(filePath));
        } catch (Exception e) {
            e.printStackTrace();
        }
        logger.info("MD5:" + hash);
    }
}